package com.yuebao.mq;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.StopWatch;
import com.alibaba.druid.support.json.JSONUtils;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yuebao.event.EventType;
import com.yuebao.exception.CustomerException;
import com.yuebao.mapper.StockMqConsumerLogMapper;

import com.yuebao.model.dto.StockMqConsumerLog;
import com.yuebao.model.dto.StockMqProducerTempLog;
import com.yuebao.request.TransactionDTO;
import com.yuebao.service.IFundCommonService;
import com.yuebao.service.impl.RedissonService;
import com.yuebao.utils.JsonUtils;
import lombok.extern.slf4j.Slf4j;

import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ExecutorService;

import static com.yuebao.event.EventType.*;

@Component
@Slf4j
public class EventMqReceiver {
    
    @Resource(name = "messageConsumeExecutor")
    private ExecutorService messageConsumeExecutor;
    
    @Resource
    private StockMqConsumerLogMapper stockMqConsumerLogMapper;
    
    @Resource
    private RedissonService redissonService;
    
    @Resource
    private IFundCommonService fundCommonService;
    
    ObjectMapper objectMapper=new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    
    public void handleReceiveMessage(String eventString, String messageId) throws Exception {
    
        
        EventDTO eventDTO = objectMapper.readValue(eventString, EventDTO.class);
        
        //防止重复消费
        if (redissonService.isMessageExist(messageId)){
            return ; //正在消费
        }else {
            //查库是否消费过
            if(this.isMessageConsumed(messageId)) {
                log.info(messageId+"消费过");
                return;
            }
        }
        log.info("提交消息消费任务");
        //多线程消费 防止积压
       
        messageConsumeExecutor.submit(()->{
            StopWatch stopWatch=new StopWatch(messageId);
            // 消费消息
            stopWatch.start();
            this.handleMessage(messageId, eventDTO);
            stopWatch.stop();
            log.info("消息消费耗时情况===> 【{}】", stopWatch);
      });
    
    
    }
    
    private boolean isMessageConsumed(String messageId) {
        QueryWrapper<StockMqConsumerLog> queryWrapper=new QueryWrapper<>();
        queryWrapper.eq("message_id", messageId);
        return !Objects.isNull(this.stockMqConsumerLogMapper.selectOne(queryWrapper));
    }
    
    private void handleMessage(String messageId, EventDTO eventDTO)  {
        
        
        try{
            //消费加锁（redis）
            if(!redissonService.messageTryLock(messageId)){
                log.error("消息加锁失败");
                return;
            }
            log.info("开始消费消息{}打印记录",JSONUtils.toJSONString(eventDTO.getMsg()));
            
            //保存消费记录
            this.insertToStockMqConsumerLog(messageId,eventDTO);
            log.info("保存成功");
            
            eventDispatcher(eventDTO);
            
        }catch (Exception e){
            log.error("消费出错：{}",e.getMessage());
            
        } finally {
            redissonService.messageIdUnLock(messageId);
        }
    }
    
    private void insertToStockMqConsumerLog(String messageId,EventDTO eventDTO) {
        StockMqConsumerLog stockConsumerLog = StockMqConsumerLog.builder()
                .createTime(DateTime.now())
                .groupName("YUEBAO")
                .msgBody(JsonUtils.writeValueAsString(eventDTO))
                .msgTopic("fund-service")
                .msgStatus(1)
                .messageId(messageId)
                .build();
        stockMqConsumerLogMapper.insert(stockConsumerLog);
    
    }
    
    private void eventDispatcher(EventDTO eventDTO){
        EventType type = EventType.fromValue(eventDTO.getType());
        try{
            switch (type){
                case REDUCE_FUND_STOCK:
                    log.info("开始处理库存扣减事件");
                    TransactionDTO transactionDTO = objectMapper.readValue(eventDTO.getMsg(), TransactionDTO.class);
                    Boolean result = fundCommonService.reduceFundShare(transactionDTO);
                    //  失败要回滚
                    break;
                default:
                    log.error("eventDTO{}没有匹配到事件",eventDTO);
                    break;
            }
        }catch (Exception e){
            
            log.error("{}",e.getMessage());
        }
        
    }
    
}
